package com.atguigu.edu.realtime.app.dwd;

import com.atguigu.edu.realtime.Base.BaseSQLApp;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.utils.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().init(3003,2,"DwdTradeOrderDetail");
    }

    /**
    * 交易域下单事务事实表
    *   order_info
    *   order_detail
     *   dwd_traffic_page 页面信息拿来源字段
    * */
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //设置ttl
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        //读取ods_db
        readOdsDb(tEnv,"DwdTradeOrderDetail");

        //读取dwd_traffic_page数据
        tEnv.executeSql("create table dwd_traffic_page(" +
                "`common` map<string,string>," +
                "`page` map<string,string>," +
                "`ts` bigint," +
                "`pt` as proctime()," +
                "`et` as to_timestamp_ltz(ts, 0)," +
                " watermark for `et` as `et` - interval '3' second " +
                ")" + SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRAFFIC_PAGE,"DwdTradeOrderDetail1","json"));

        //从dwd_traffic_page中拿到source_id和mid(独立访客数标识)
        Table orderSource = tEnv.sqlQuery("select " +
                "`common`['sc'] source_id," +
                "`common`['sid'] session_id " +
                "from dwd_traffic_page ");
        tEnv.createTemporaryView("order_source",orderSource);

        //从ods_db中过滤order_detail
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['course_id'] course_id," +
                "data['course_name'] course_name," +
                "data['order_id'] order_id," +
                "data['create_time'] create_time," +
                "data['origin_amount'] origin_amount," +
                "data['coupon_reduce'] coupon_reduce," +
                "data['final_amount'] final_amount," +
                "ts," +
                "pt " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail",orderDetail);

        //从ods_db中过滤order_info
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['province_id'] province_id," +
                "data['session_id'] session_id " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and `type`='insert'");
        tEnv.createTemporaryView("order_info",orderInfo);

        //join
        Table result = tEnv.sqlQuery("select " +
                "od.id," +
                "od.order_id," +
                "oi.user_id," +
                "od.course_id," +
                "od.course_name," +
                "oi.province_id," +
                "os.source_id," +
                "od.origin_amount," +
                "od.coupon_reduce," +
                "od.final_amount," +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id," +
                "od.ts " +
                "from order_detail od " +
                "join order_info oi on od.order_id=oi.id " +
                "join order_source os on oi.session_id=os.session_id");

        //把结果写出到kafka
        tEnv.executeSql("create table dwd_trade_order_detail(" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "course_id string," +
                "course_name string," +
                "province_id string," +
                "source_id string," +
                "origin_amount string," +
                "coupon_reduce string," +
                "final_amount string," +
                "date_id string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "json"));
       result.executeInsert("dwd_trade_order_detail");
    }
}
